package com.markyao.recos.raft.core;


import com.markyao.recos.raft.raftRole.AbstractRaftRole;
import com.markyao.recos.raft.raftRole.RaftRoleImpl;
import com.markyao.recos.regist.RecosRegistConfig;
import com.markyao.timer.TimerWorker;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j(topic = "m.RaftCore")
@DependsOn({"timerWorker","authController","raftController","raftBeanMapsConfig"})
public class RaftCore {
    private final ScheduledExecutorService scheduledExecutorService0=new ScheduledThreadPoolExecutor(1,new DefaultThreadFactory("raftHeartBeat0"));
    private final ScheduledExecutorService scheduledExecutorService1=new ScheduledThreadPoolExecutor(1,new DefaultThreadFactory("raftHeartBeat1"));
    @Autowired
    private TimerWorker timerWorker;
    @Autowired
    private RecosRegistConfig recosRegistConfig;
    @Autowired
    @Qualifier("raftHearBeat")
    AtomicInteger raftHearBeat;

    @Autowired
    @Qualifier("raftRole")
    private AbstractRaftRole raftRole;
    @Autowired
    @Qualifier("isElection")
    private volatile AtomicInteger isElection;

    @Autowired
    @Qualifier("leaderMetadata")
    ConcurrentHashMap<String,Object> leaderMetadata;



    private int randomWaitMills;
    private long startMills;
    private int heartListenT=350;

    @PostConstruct
    public void init(){
        startMills=timerWorker.getMills();
        Random random=new Random();
        int rd = random.nextInt(4);
        randomWaitMills=150+rd*50;//150ms ~ 300ms
        log.info("【Recos集群模式启动】初始化开始时间为: {} 等待周期为: {}ms",startMills,randomWaitMills);

        new Thread(()->{
            if (raftRole.isOnload()){
                //需要先与各方建立连接,交换token
                log.info("trying to connect to every node...");
                raftRole.connenct2Recos();
                if (!raftRole.isOnload()){
                    // TODO: 2023/9/16 等待接收心跳,如果过一段时间没有收到心跳,那么将成为候选人
                    scheduledExecutorService0.scheduleAtFixedRate(new RaftHeartBeatListener(),startMills,randomWaitMills, TimeUnit.MILLISECONDS);
                    scheduledExecutorService1.scheduleAtFixedRate(new RaftHeartBeatClearner(),startMills,heartListenT, TimeUnit.MILLISECONDS);
                }
            }
        }).start();

    }


    class RaftHeartBeatListener implements Runnable{
        private int retryHeart=0;

        @Override
        public void run() {
            try {
                raftListener();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private void raftListener() {
            if (raftRole.isOnload()){
                return;
            }
            if (isElection.get()==1){
                //竞选周期
                if (raftRole.getRoleState()==1){
                    if (raftHearBeat.get()==1){
                        //结束竞选周期
                        log.info("竞选周期收到心跳,结束周期");
                        isElection.set(0);
                    }
                }else if (raftRole.getRoleState()==2){

                }else {
                    isElection.set(0);
                }
            }else {
                //非竞选周期
                if (raftRole.getRoleState()==1){
                    if (raftHearBeat.get()==1 || isElection.get()==1){
                        retryHeart=0;
                        return;
                    }
                    //如果没有收到心跳,那么将转化为候选人，发起选举,开始竞选周期
                    if (retryHeart<3 && raftHearBeat.get()==0 && isElection.get()!=1){
                        retryHeart++;
                        return;
                    }
                    boolean alive = (boolean) leaderMetadata.get("alive");
                    if (raftHearBeat.get()==0){
                        if (!raftRole.isBoom() && !alive && !raftRole.isVote() && raftRole.getRoleState()==1  && isElection.get()!=1){
                            isElection.compareAndSet(0,1);
                            raftRole.shiftCandidate();
                            raftRole.election();
                        }
                    }
                }else if (raftRole.getRoleState()==2){
                    log.error("非竞选周期，没有候选人节点，系统出错");
                }else {
                    leaderMetadata.put("host",raftRole.getHost());
                    leaderMetadata.put("port",raftRole.getPort());
                    leaderMetadata.put("alive",true);
                    recosRegistConfig.setLeader(true);
                }
            }
        }
    }
    class RaftHeartBeatClearner implements Runnable{
        private AtomicInteger noHeartBeat;
        private int askT;
        private double askSec;
        public RaftHeartBeatClearner() {
            noHeartBeat=new AtomicInteger(0);
            askT=3+new Random().nextInt(4);
            askSec=askT*heartListenT;
        }

        //定期检查并 清理心跳
        @Override
        public void run() {
            if (raftRole.isOnload()){
                return;
            }
            if (raftRole.getRoleState()==1){
                if (raftHearBeat.get()==0){
                    noHeartBeat.incrementAndGet();
                    if (noHeartBeat.get()>=askT){
                        ask2Leader();
                    }

                }else {
                    noHeartBeat.set(0);
                }
                raftHearBeat.compareAndSet(1,0);
            }
        }

        private void ask2Leader() {
            log.error("超过{}ms没有心跳了,向领导人发起询问.",askSec);
            noHeartBeat.set(0);
            String host = leaderMetadata.get("host").toString();
            int port= (int) leaderMetadata.get("port");
            boolean f=raftRole.askLeader(host,port);
            if (!f){
                raftRole.setVote(false);
                leaderMetadata.put("alive",false);
            }else {
                raftRole.setVote(true);
                leaderMetadata.put("alive",true);
            }
        }
    }

}
